/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.mem.impl;

import org.sabayframework.activity.ActivityContext;
import org.sabayframework.activity.impl.ReadWriteActivity;
import org.sabayframework.channel.Lock;
import org.sabayframework.mem.MemException;
import org.sabayframework.mem.Memorizable;
import org.sabayframework.mem.Transaction.Status;
import org.sabayframework.mem.TransactionCompletedException;

public abstract class TxActivity extends ReadWriteActivity<Object, Object, TxRequest> {
	
	private final AbstractSpace space;
	private final TransactionImpl transaction;
	
	protected TxActivity(TransactionImpl transaction, AbstractSpace tm) {
		super();
		this.transaction = transaction;
		this.space = tm;
	}

	public AbstractSpace getSpace() {
		return space;
	}

	public TransactionImpl getTransaction() {
		return transaction;
	}

	protected abstract Boolean[] contains(long[] ids) throws Exception;
	protected abstract Memorizable[] read(Class<? extends Memorizable> objectClass, long[] ids)throws Exception;
	protected abstract boolean write(Memorizable[] r)throws Exception;
	protected abstract boolean delete(long[] ids)throws Exception;
	protected abstract boolean complete(Status status)throws Exception;
	
	@Override
	public void read(ActivityContext<Object, Object> ctx, Lock<Object> inLock,
			Lock<Object> outLock, TxRequest param) throws Exception {
		write(ctx, inLock, outLock, param);
	}

	@Override
	public void write(ActivityContext<Object, Object> ctx, Lock<Object> iLock,
			Lock<Object> oLock, TxRequest param) throws Exception {
		final Status status = transaction.getStatus();
		if(status != Status.ACTIVE) {
			Throwable e=new TransactionCompletedException();
			iLock.cancel();
			oLock.cancel();
			ctx.getInput().fail(e);
			ctx.getOutput().fail(e);
			return;
		}
		try{
			internalExecute(param, iLock, oLock);
		} catch (Throwable e) {
			ctx.getInput().fail(e);
			ctx.getOutput().fail(e);
			oLock.cancel();
		}
	}

	private void internalExecute(TxRequest r, Lock<Object> iLock, Lock<Object> oLock) throws Exception {
		if(space.isClosed()) throw new IllegalStateException("Space is closed");
		final int size = iLock.size();
		TxRequest.Type type = r.type;
		switch(type){
		case COMMIT:
			if(complete(Status.COMMIT)){
				transaction.setStatus(Status.COMMIT);
				space.completed(transaction);
				oLock.cancel();
			}else{
				iLock.cancel();
				oLock.cancel();
				return;
			}
			break;
		case ROLLBACK:
			if(complete(Status.ROLLBACK)){
				transaction.setStatus(Status.ROLLBACK);
				space.completed(transaction);
				oLock.cancel();
			}else{
				iLock.cancel();
				oLock.cancel();
				return;
			}
			break;
		case DELETE:
			{
				long[] ids = new long[size];
				for(int i=0;i<size;++i){
					ids[i] = ((Long)iLock.get(i)).longValue();
				}
				if(!delete(ids)){
					iLock.cancel();
				}
				oLock.cancel();
			}
			break;
		case CONTAINS:
			{
				long[] ids = new long[size];
				for(int i=0;i<size;++i){
					ids[i] = ((Long)iLock.get(i)).longValue();
				}
				Boolean[] rs = contains(ids);
				if(rs!=null){
					for(int i=0;i<size;++i)	oLock.set(i, rs[i]);
				}else{
					iLock.cancel();
					oLock.cancel();
					return;
				}
			}
			break;
		case READ:
			{
				long[] ids = new long[size];
				for(int i=0;i<size;++i){
					ids[i] = ((Long)iLock.get(i)).longValue();
				}
				Class<? extends Memorizable> objectClass = r.objectClass;
				Memorizable[] rs = read(objectClass, ids);
				if(rs!=null){
					for(int i=0;i<size;++i)	oLock.set(i, rs[i]);
				}else{
					iLock.cancel();
					oLock.cancel();
					return;
				}
			}
			break;
		case WRITE:
			{	
				Memorizable[] rs = new Memorizable[size];
				for(int i=0;i<size;++i){
					rs[i]=(Memorizable)iLock.get(i);
				}
				if(!write(rs)){
					iLock.cancel();
				}
				oLock.cancel();
			}
			break;
		default:
			throw new MemException("Invalid transaction type: "+type);
		}
	}
}

